package com.atguigu.gmall.realtime.app.demo;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class FlinkApiJoinApp {


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        String leftTopic="demo_left_topic";
        String rightTopic="demo_right_topic";

        String groupId="demo_join_app";


        DataStreamSource<String> leftStream = env.addSource(MyKafkaUtil.getKafkaConsumer(leftTopic, groupId));
        DataStreamSource<String> rightStream = env.addSource(MyKafkaUtil.getKafkaConsumer(rightTopic, groupId));

        SingleOutputStreamOperator<JSONObject> leftJsonStream = leftStream.map(JSON::parseObject);
        SingleOutputStreamOperator<JSONObject> rightJsonStream = rightStream.map(JSON::parseObject);
       // 分配watermark
        SingleOutputStreamOperator<JSONObject> leftJsonWmStream = leftJsonStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofMillis(2000))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long recordTimestamp) {
                        return jsonObject.getLong("ts");
                    }
                }));


        SingleOutputStreamOperator<JSONObject> rightJsonWmStream = rightJsonStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long recordTimestamp) {
                        return jsonObject.getLong("ts");
                    }
                }));
        // window join

        //订单  {"id":"101","user_id":"u_101","status":"1001", "amount":200,"ts":1000}
        // 订单明细 {"id":"990","order_id":"101","sku_id":"s_77",  "ts":1000}
        //inner join
//        DataStream<JSONObject> joinedStream = leftJsonWmStream.join(rightJsonWmStream)
//                .where(leftJson -> leftJson.getString("id"))
//                .equalTo(rightJson -> rightJson.getString("order_id"))
//                .window(TumblingEventTimeWindows.of(Time.milliseconds(10000)))  // 订单和订单明细 事件时间相同  不会出现在不同窗口上 所以可以用滚动窗口
//                .apply(new JoinFunction<JSONObject, JSONObject, JSONObject>() {
//                    @Override
//                    public JSONObject join(JSONObject leftJson, JSONObject rightJson) throws Exception {
//                        JSONObject joinedJson = new JSONObject();
//                        joinedJson.put("detail_id", rightJson.getString("id"));
//                        joinedJson.put("order_id", leftJson.getString("id"));
//                        joinedJson.put("user_id", leftJson.getString("user_id"));
//                        joinedJson.put("status", leftJson.getString("status"));
//                        joinedJson.put("sku_id", rightJson.getString("sku_id"));
//                        joinedJson.put("ts", rightJson.getLong("ts"));
//                        return joinedJson;
//                    }
//                });
        //left join
        //  订单明细
        //   {"id":"991","order_id":"101","amount":200,"ts":1000}
        //   {"id":"992","order_id":"102","amount":400,"ts":1000}
        //   {"id":"993","order_id":"103","amount":100,"ts":11000}
        //
        //   购物券使用
        //   {"id":"7701","order_detail_id":"991","coupon_id":"3301" ,"ts":1000}
        //   {"id":"7702","order_detail_id":"993","coupon_id":"3302" ,"ts":12000}
//        DataStream<JSONObject> leftJoinWindowStream = leftJsonWmStream.coGroup(rightJsonWmStream)
//                .where(leftJson -> leftJson.getString("id"))
//                .equalTo(rightJson -> rightJson.getString("order_detail_id"))
//                .window(TumblingEventTimeWindows.of(Time.milliseconds(10000)))
//                .apply(new CoGroupFunction<JSONObject, JSONObject, JSONObject>() {
//                    @Override
//                    public void coGroup(Iterable<JSONObject> leftItr, Iterable<JSONObject> rightItr, Collector<JSONObject> out) throws Exception {
//                        for (JSONObject leftJson : leftItr) {
//                            boolean ifExistsRight = false;
//                            for (JSONObject rightJson : rightItr) {   //如果右边有数据则合并到左边 并且输出
//                                ifExistsRight = true;
//                                leftJson.put("coupon_id", rightJson);
//                                out.collect(leftJson);
//                            }
//                            if (!ifExistsRight) {       //如果没有 左边独自输出
//                                out.collect(leftJson);
//                            }
//                        }
//                    }
//                });

       // leftJoinWindowStream.print(" window left join  :::");

        //订单：
        //{"id":"101","user_id":"u_101","status":"1001", "amount":200,"ts":1000}
        //{"id":"102","user_id":"u_102","status":"1001", "amount":400,"ts":11000}
        //{"id":"103","user_id":"u_103","status":"1001", "amount":600,"ts":15000}


        //支付
        //  {"id":"55","order_id":"101","type":"wx"  ,"ts":4000}
        //  {"id":"56","order_id":"102","type":"ali" ,"ts":12000}

        KeyedStream<JSONObject, String> leftKeyedStream = leftJsonWmStream.keyBy(leftjson -> leftjson.getString("id"));
        KeyedStream<JSONObject, String> rightKeyedStream = rightJsonWmStream.keyBy(rightjson -> rightjson.getString("order_id"));

        SingleOutputStreamOperator<JSONObject> intervalJoinedStream = leftKeyedStream.intervalJoin(rightKeyedStream)
                .between(Time.milliseconds(-50), Time.milliseconds(60000 * 15 + 10000))
                .process(new ProcessJoinFunction<JSONObject, JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject left, JSONObject right, ProcessJoinFunction<JSONObject, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        left.put("type", right.getString("type"));
                        left.put("pay_ts", right.getLong("ts"));
                        out.collect(left);
                    }
                });
        intervalJoinedStream.print("interval join :::");

        env.execute();
    }
}
